package com.spark.core.examples

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object TopN1 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("topN")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val infos = sc.textFile("./data/scores.txt", 10)

    val pairInfo: RDD[(String, Int)] = infos.map(one => {
      (one.split("\t")(0), one.split("\t")(1).toInt)
    })

    val classScoreRdd: RDD[(String, Iterable[Int])] = pairInfo.groupByKey()

    val result: RDD[(String, List[Int])] = classScoreRdd.map(kv => {
      val className: String = kv._1
      val scoreIter: Iterable[Int] = kv._2
      val scoreList: List[Int] = scoreIter.toList.sortWith((x1, x2) => {
        x1.compare(x2) > 0
      })
      (className, scoreList.take(3))
    })

    result.foreach(println)


  }

}
